package com.atguigu.day07;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink08_Process {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);

        //3.将数据按照空格切出一个个单词
        SingleOutputStreamOperator<String> wordsStream = streamSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
//                ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime()+2000);
                String[] split = value.split(" ");
                for (String s : split) {
                    out.collect(s);
                }
            }

          /*  @Override
            public void onTimer(long timestamp, ProcessFunction<String, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                System.out.println("定时器触发");
            }*/
        });
        //4.将每个单词组成Tuple2元组
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordToOneStream = wordsStream.process(new ProcessFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void processElement(String value, ProcessFunction<String, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                out.collect(Tuple2.of(value, 1));
            }
        });
                /*.process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public void processElement(Tuple2<String, Integer> value, ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                        if (!value.f0.equals("hello")){
                            out.collect(value);
                        }
                    }
                })
                ;*/

        //5.将相同单词的数据聚合到一块
        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = wordToOneStream.keyBy("f0");

        //6.对单词计数  实现count功能
        keyedStream.process(new KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Integer>>() {


            //定义累加器
//            HashMap<String, Integer> sum = new HashMap<>();
            ReducingState<Integer> reducingState;

            @Override
            public void open(Configuration parameters) throws Exception {
                reducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Integer>("sum1", new ReduceFunction<Integer>() {
                    @Override
                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                        return value1 + value2;
                    }
                }, Integer.class));
            }

            @Override
            public void processElement(Tuple2<String, Integer> value, KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
//                //先判断这个单词是否计过数
//                if (sum.containsKey(value.f0)){
//                    //有K 证明不是第一次来
//                    //获取之前计数的结果
//                    Integer lastSum = sum.get(value.f0);
//                    //更新计数结果
//                    sum.put(value.f0, lastSum + 1);
//                }else {
//                    //没 K 证明第一次来
//                    sum.put(value.f0, 1);
//                }

                reducingState.add(value.f1);

//                out.collect(Tuple2.of(value.f0,sum.get(value.f0)));
                out.collect(Tuple2.of(value.f0,reducingState.get()));
            }
        }).print();

        env.execute();
    }
}
